/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package replica.communication;

import java.io.Serializable;
import java.util.Iterator;
import java.util.List;
import java.util.Properties;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.jms.*;
import javax.naming.InitialContext;
import replica.transaction.ReplicaManager;

/**
 *
 * @author Elena
 */
public class MessageDispatcherReplicaHost implements MessageDispatcher,javax.jms.MessageListener{
    
    TopicConnection connection = null;
    List<ReplicaManager> managers;
 

    
    public MessageDispatcherReplicaHost(List<ReplicaManager> managers) 
    {
       this.managers =managers; 
       //System.out.println("Sono MessageDispatcherReplicaHost ");
       try{
              buildSubscriber();}
       catch(Exception e){
           e.printStackTrace();
           System.exit(-1007);
       }
    }

  
   public void buildSubscriber() throws Exception{
     
        try {
                       
                    Properties env = new Properties( );
                    InitialContext jndi = new InitialContext(env);

                    
                    TopicConnectionFactory conFactory =
                    (TopicConnectionFactory)jndi.lookup("jms/myConnectionFactory");
                    
                    TopicConnection  connection = conFactory.createTopicConnection();
                    
                    TopicSession subSession = connection.createTopicSession(false,
                                                Session.AUTO_ACKNOWLEDGE);

                    
                    Topic Topic = (Topic)jndi.lookup("jms/Topic");
                    TopicSubscriber subscriber = subSession.createSubscriber(Topic);
                    
                    subscriber.setMessageListener(this);
                    connection.start( );
                
                System.out.println("MessageDispatcherReplicaHost: connectionStarted");

    } catch (Exception e) {
            e.printStackTrace();System.exit(-8742);
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                }
            }
        }
    }

   
   
    @Override
     public void onMessage(Message message) {
        System.out.println(  "Sono OnMessage ");
        try {
            ObjectMessage objectmessage= (ObjectMessage) message;
            Serializable extracted=objectmessage.getObject();
            Messaggio mes = (Messaggio)extracted;
            long nodeId = mes.getNodeId();
            System.out.println("MDReplicaHost  Read message: "+mes+" from front end "
                    +nodeId);
            
           
            for(ReplicaManager r: managers){
                    if(nodeId==r.getNodeId())
                    {
                        //r.consumeMessages(mes.status());
                        r.add(mes.status());
                    }
                    
                    
                 }
               
        } catch (Exception jmse)
        { jmse.printStackTrace( ); 
            System.exit(-237654);
        }
    }
    

    
}
